package simpledb;

import java.util.*;

/**
 * The Aggregator operator that computes an aggregate (e.g., sum, avg, max,
 * min).  Note that we only support aggregates over a single column, grouped
 * by a single column.
 */
public class Aggregate extends AbstractDbIterator {


    DbIterator dbIterator ;

    Aggregator aggregator ;

    DbIterator realDbIterator ;

    int gbfield ;
    int afield ;
    Aggregator.Op what ;


    /**
     * Constructor.  
     *
     *  Implementation hint: depending on the type of afield, you will want to construct an 
     *  IntAggregator or StringAggregator to help you with your implementation of readNext().
     * 
     *
     * @param child The DbIterator that is feeding us tuples.
     * @param afield The column over which we are computing an aggregate.
     * @param gfield The column over which we are grouping the result, or -1 if there is no grouping
     * @param aop The aggregation operator to use
     */
    public Aggregate(DbIterator child, int afield, int gfield, Aggregator.Op aop) {
        // some code goes here

        this.dbIterator = child ;
        this.afield = afield ;
        this.gbfield = gfield ;
        this.what = aop ;
    }

    public static String aggName(Aggregator.Op aop) {
        switch (aop) {
        case MIN:
            return "min";
        case MAX:
            return "max";
        case AVG:
            return "avg";
        case SUM:
            return "sum";
        case COUNT:
            return "count";
        }
        return "";
    }

    public void open()
        throws NoSuchElementException, DbException, TransactionAbortedException {
        // some code goes here

        dbIterator.open();

        TupleDesc tupleDesc = dbIterator.getTupleDesc() ;

        Type aFieldType = tupleDesc.getType( afield );
        Type gbFieldType = null ;

        if( gbfield != -1 ){
            gbFieldType = tupleDesc.getType( gbfield );
        }

        System.out.println( " :::: tupleDesc.getType( afield ); ::::: " + aFieldType  );

        if( aFieldType.equals( Type.INT_TYPE ) ){

            if( gbFieldType == null || gbFieldType.equals( Type.INT_TYPE ) ){
                aggregator  = new  IntAggregator(   gbfield , Type.INT_TYPE ,   afield, what ) ;
            }else if( gbFieldType.equals( Type.STRING_TYPE ) ){
                aggregator  = new SumStringAggregator(   gbfield , gbFieldType ,   afield, what ) ;
            }
        }else {
            aggregator  = new  StringAggregator(   gbfield , Type.INT_TYPE ,   afield, what ) ;

        }

        while ( dbIterator.hasNext() ){
            aggregator.merge( dbIterator.next() );
        }

        realDbIterator  = aggregator.iterator();
    }

    /**
     * Returns the next tuple.  If there is a group by field, then 
     * the first field is the field by which we are
     * grouping, and the second field is the result of computing the aggregate,
     * If there is no group by field, then the result tuple should contain
     * one field representing the result of the aggregate.
     * Should return null if there are no more tuples.
     */
    protected Tuple readNext() throws TransactionAbortedException, DbException {
        // some code goes here

        if( realDbIterator.hasNext() ){

            Tuple tuple = realDbIterator.next();
            Tuple newTuple  = null ;

            if( gbfield == -1 ){

                newTuple  = new Tuple( Utility.getTupleDesc(1) );
                newTuple.setField(0,  tuple.getField(0)   );
            }else{

                Type gbfieldType = tuple.getTupleDesc().getType( gbfield );

                if(  gbfieldType.equals( Type.STRING_TYPE ) ){
                    TupleDesc tupleDesc = new TupleDesc( new Type[]{ Type.STRING_TYPE, Type.INT_TYPE });
                    newTuple  = new Tuple(  tupleDesc );
                }else{
                    newTuple  = new Tuple( Utility.getTupleDesc(2)  );
                }
                newTuple.setField(0,  tuple.getField(0)   );
                newTuple.setField(1, tuple.getField(1) );
            }

            return newTuple ;
        }

        return null;
    }

    public void rewind() throws DbException, TransactionAbortedException {
        // some code goes here

        realDbIterator.rewind();
    }

    /**
     * Returns the TupleDesc of this Aggregate.
     * If there is no group by field, this will have one field - the aggregate column.
     * If there is a group by field, the first field will be the group by field, and the second
     * will be the aggregate value column.
     * 
     * The name of an aggregate column should be informative.  For example:
     * "aggName(aop) (child_td.getFieldName(afield))"
     * where aop and afield are given in the constructor, and child_td is the TupleDesc
     * of the child iterator. 
     */
    public TupleDesc getTupleDesc() {
        // some code goes here

        System.out.println( "===  Aggregate  getTupleDesc  ====" );
        String aggName =  aggName( what );
        String fieldName = dbIterator.getTupleDesc().getFieldName( afield );

        if( this.gbfield ==  -1){
//            no group
           return  Utility.getTupleDesc(1);
        }else{
            return Utility.getTupleDesc(2);
        }

    }

    public void close() {
        // some code goes here

        dbIterator.close();
        realDbIterator.close();
    }


}
